package cn.dmp.charts.arealDistribution

import cn.dmp.utils.ParseUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * create by lfq on 171106
  * wordcount
  */
object ArealDistribution171106V3 {
  def main(args: Array[String]): Unit = {

    if (args.length != 1) {
      println(
        """
          |cn.dmp.report.ProCityRpt
          |参数：
          | logInputPath
        """.stripMargin)
      sys.exit()
    }

    val Array(logInputPath) = args

    val conf: SparkConf = new SparkConf().setAppName("ArealDistribution171106V3").setMaster("local[4]")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .set("spark.sql.parquet.compression.codec", "snappy")

    val sc: SparkContext = new SparkContext(conf)

    val rawRDD: RDD[String] = sc.textFile(logInputPath)

    val splitsRDD: RDD[Array[String]] = rawRDD.map(_.split(",",-1))

    //过滤
    val suitedLengthRDD = splitsRDD.filter(_.length == 85)

    //取出判断条件字段
    val baseRDD = suitedLengthRDD.map(splits => {

      val provincename: String = splits(24)
      val cityname: String = splits(25)
      val requestmode: Int = ParseUtils.parseInt(splits(8))
      val processnode: Int = ParseUtils.parseInt(splits(35))
      val iseffective: Int = ParseUtils.parseInt(splits(30))
      val isbilling: Int = ParseUtils.parseInt(splits(31))
      val isbid: Int = ParseUtils.parseInt(splits(39))
      val iswin: Int = ParseUtils.parseInt(splits(42))
      val adorderid: Int = ParseUtils.parseInt(splits(2))
      val winprice: Double = ParseUtils.parseDouble(splits(41))
      val adpayment: Double = ParseUtils.parseDouble(splits(75))

      (requestmode, processnode, iseffective, isbilling, isbid, iswin, adorderid, winprice, adpayment,provincename, cityname)

    })

    //对数据进行处理，如果满足业务逻辑条件，值为1，否则为0.对处理完的数据reduceByKey

    val adMatchingRDD: RDD[((String, String), (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = baseRDD.map(t => {
      var primaryRequest = 0

      if (t._1 == 1 && t._2 >= 1) {
        primaryRequest = 1
      }

      var effectiveRequest = 0
      if (t._1 == 1 && t._2 >= 2) {
         effectiveRequest = 1
      }

      var AdRequest = 0
      if (t._1 == 1 && t._2 == 3) {
         AdRequest = 1
      }

      var biddingTimes = 0
      if (t._3 == 1 && t._4 == 1 && t._5 == 1 && t._7 != 0) {
        biddingTimes = 1
      }

      var succesedBiddingTime = 0
      if (t._3 == 1 && t._4 == 1 && t._6 == 1) {
         succesedBiddingTime = 1
      }

      var showTimes = 0
      if (t._1 == 2 && t._3 == 1) {
        showTimes = 1
      }

      var clickTimes = 0
      if (t._1 == 3 && t._3 == 1) {
        clickTimes = 1
      }

      var AdConsume= 0.0
      if (t._3 == 1 && t._4 == 1 && t._6 == 1) {
         AdConsume = t._8 / 1000
      }

      var AdCost = 0.0
      if (t._3 == 1 && t._4 == 1 && t._6 == 1) {
       AdCost = t._9 / 1000
      }

      val province = t._10
      val city = t._11

      ((province, city), (primaryRequest, effectiveRequest, AdRequest, biddingTimes, succesedBiddingTime, showTimes, clickTimes, AdConsume, AdCost))

    })

    //聚合
    val reducedRDD: RDD[((String, String), (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = adMatchingRDD.reduceByKey((t1, t2) => {
      ((t1._1 + t2._1), (t1._2 + t2._2), (t1._3 + t2._3), (t1._4 + t2._4), (t1._5 + t2._5), (t1._6 + t2._6), (t1._7 + t2._7), (t1._8 + t2._8), (t1._9 + t2._9))
    })


    reducedRDD.foreach(x=>{
      println(x)
    })
    







  }
}
